package tom.vertx.core.demo11;

import io.vertx.core.AbstractVerticle;
import io.vertx.core.Promise;
import io.vertx.core.Vertx;
import io.vertx.core.file.OpenOptions;
import io.vertx.core.net.NetServer;
import io.vertx.core.net.NetServerOptions;
import io.vertx.core.streams.Pump;

/**
 * @author ZHUFEIFEI
 */
public class StreamDemo {

  public static void main(String[] args) {
    Vertx.vertx().deployVerticle(new V());
  }

  static class V extends AbstractVerticle {
    @Override
    public void start(Promise<Void> startPromise) {
      NetServer server = vertx.createNetServer(
        new NetServerOptions().setPort(1234).setHost("localhost")
      );

      server.connectHandler(sock -> {
        //
        sock.handler(buffer -> {
          //socket实现了ReadStream WriteStream, 此处写数据，会立即返回，数据放入队列，当队列满时，不再写入
          sock.write(buffer);
          //如果写队里列满了，暂停，ReadStream WriteStream是采用背压back-press模型
          if (sock.writeQueueFull()) {
            //立即返回
            sock.pause();
            //当队列恢复未满状态时，会调用drainHandler,恢复sock即可继续接收写入
            sock.drainHandler(done -> {
              sock.resume();
            });
          }
          //上面的写入和队列判断操作，都可以通过pipeTo实现，流到流，等同于上述的操作， pipeTo可以写到任意实现了WriteStream的流，如FileSystem
          //sock.pipeTo(sock);
          //Pump从readStream 写入到 writeStream， NetSocket实现了ReadStream WriteStream
          //Pump.pump(sock, sock);
        });

      }).listen();

      vertx.fileSystem().open("D:/log/abc", new OpenOptions(), res -> {
          if (res.succeeded()) {
            //从一个文件读取写入到另一个文件
            Pump.pump(res.result(), vertx.fileSystem().openBlocking("D:/log/aaa.pump", new OpenOptions().setCreate(true)))
              .start();
            System.out.println("pump file start.");
          } else {
            System.out.println("open file failed => " + res.cause().toString());
          }
      });

      startPromise.complete();
    }
  }
}
